package com.els.pool;

import java.util.List;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import org.apache.log4j.Logger;

import com.els.BulkPool;
import com.els.ElsConnection;
import com.els.ElsPool;
import com.els.connection.ConfigElastic;
import com.google.gson.JsonObject;

public class ConfigElasticPool implements ElsPool,BulkPool{
	Logger logger=Logger.getLogger(this.getClass());
	private String configPath;
	private Queue<ElsConnection> pool=new ConcurrentLinkedQueue<ElsConnection>();
	private int poolSize=10;
	private final int _buffer=10000;
	private boolean hasInit=false;
	public ConfigElasticPool(String configPath){
		this.configPath=configPath;
	}
	public void initPool(){
		if(hasInit)return;	//如果已经初始化过，则直接返回
		for(int i=0;i<poolSize;i++){
			ElsConnection eb=new ConfigElastic(configPath);
			eb.buildClient();
			pool.add(eb);
		}
		hasInit=true;
	}
	public ElsConnection getConnection() {
		ElsConnection ec = pool.poll();
		if(ec==null){
			 ec=new ConfigElastic(configPath);
			 ec.buildClient();
			 return ec;
		}
		return ec;
	}
	public void backConnection(ElsConnection connection) {
		if(pool.size()>poolSize){
			connection.closeClient();
		}else {
			pool.add(connection);
		}
	}
	public int getAvailableConnectionNum(){
		return pool.size();
	}
	/**
	 * @param indexName 索引名称
	 * @param typeName type名称
	 * @param id 用哪个字段作为id
	 * @param rows 索引的数据，json格式
	 * 
	 * 
	 */
	public void concurrentBulkIndex(final String indexName, final String typeName, final String id,
			final List<JsonObject> rows) {
		ExecutorService fixedThreadPool=Executors.newFixedThreadPool(poolSize);
		
		for(int fromIndex=0,toIndex=10000;
				fromIndex<rows.size();
				toIndex+=_buffer,fromIndex+=_buffer)
		{
			if(toIndex>rows.size())toIndex=rows.size();
			final List<JsonObject> tempList=rows.subList(fromIndex, toIndex);
			fixedThreadPool.execute(new Runnable() {
				public void run() {
					ElsConnection ec= getConnection();
					ec.singleBulkIndex(indexName, typeName, tempList, id);
					backConnection(ec);
				}
			});
		}
		fixedThreadPool.shutdown();
		try {
			fixedThreadPool.awaitTermination(Long.MAX_VALUE, TimeUnit.DAYS);
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		logger.info("end threadpool");
		
		return;
	}
	public void concurrentBulkIndex(final String indexName,final  String typeName,
			final String id,final String parent, List<JsonObject> rows) {
		ExecutorService fixedThreadPool=Executors.newFixedThreadPool(poolSize);
		
		for(int fromIndex=0,toIndex=10000;
				fromIndex<rows.size();
				toIndex+=_buffer,fromIndex+=_buffer)
		{
			if(toIndex>rows.size())toIndex=rows.size();
			final List<JsonObject> tempList=rows.subList(fromIndex, toIndex);
			fixedThreadPool.execute(new Runnable() {
				public void run() {
					ElsConnection ec= getConnection();
					ec.singleBulkIndex(indexName, typeName, tempList, id, parent);
					backConnection(ec);
				}
			});
		}
		fixedThreadPool.shutdown();
		try {
			fixedThreadPool.awaitTermination(Long.MAX_VALUE, TimeUnit.DAYS);
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		logger.info("end threadpool");
		
		return;
	}
	
	
	public <T> void concurrentBulk(List<T> requestList){
		ExecutorService fixedThreadPool=Executors.newFixedThreadPool(poolSize);
		for(int fromIndex=0,toIndex=10000;
				fromIndex<requestList.size();
				toIndex+=_buffer,fromIndex+=_buffer)
		{
			if(toIndex>requestList.size())toIndex=requestList.size();
			final List<T> tempList=requestList.subList(fromIndex, toIndex);
			fixedThreadPool.execute(new Runnable() {
				public void run() {
					ElsConnection ec= getConnection();
					ec.singleBulk(tempList);
					backConnection(ec);
				}
			});
		}
		fixedThreadPool.shutdown();
		try {
			fixedThreadPool.awaitTermination(Long.MAX_VALUE, TimeUnit.DAYS);
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		logger.info("end threadpool");
		
		return;
	}
	public void setPoolSize(int poolSize) {
		this.poolSize=poolSize;
	}

}
